package com.yl.flink.processor;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.NumberUtil;
import com.yl.constant.Const;
import com.yl.constant.TimeUnit;
import com.yl.entity.MultiDataEntity;
import com.yl.util.DtUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayDeque;
import java.util.Optional;
import java.util.Queue;

@Slf4j
public class CalRocFunc extends KeyedProcessFunction<String, MultiDataEntity, MultiDataEntity> {

    private final Long timeSpan;
    private final Integer interval;
    private final TimeUnit timeUnit;
    private final Integer referQuotaCode;

    // 缓存一段时间内的监测数据
    private transient MapState<Integer, Queue<Tuple2<DateTime, Double>>> lastValueMap;

    public CalRocFunc(Long timeSpan, Integer interval, TimeUnit timeUnit, Integer referQuotaCode) {
        this.timeSpan = timeSpan;
        this.interval = null != interval ? interval : timeSpan.intValue();
        this.timeUnit = timeUnit;
        this.referQuotaCode = referQuotaCode;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        MapStateDescriptor<Integer, Queue<Tuple2<DateTime, Double>>> stateDescriptor = new MapStateDescriptor<>(
                Const.S_CACHE_RECEIVE,
                BasicTypeInfo.INT_TYPE_INFO,
                TypeInformation.of(new TypeHint<>() {
                }));
        lastValueMap = getRuntimeContext().getMapState(stateDescriptor);
    }

    /**
     * 计算变化速率
     */
    @Override
    public void processElement(MultiDataEntity multiDataEntity, KeyedProcessFunction<String, MultiDataEntity, MultiDataEntity>.Context context, Collector<MultiDataEntity> out) throws Exception {
        // 更新缓存中的数据
        cacheLastValue(multiDataEntity);
        Tuple2<Integer, Double> calValues = multiDataEntity.getCalValues();
        // 测点指标数据
        Double thisValue = multiDataEntity.getRawValues().get(referQuotaCode);
        // 获取上次监测值对象
        Double lastValue = getLastValue(multiDataEntity.getTarget().getId());
        if (null != lastValue && interval != 0) {
            // 计算变化速率
            Double roc = NumberUtil.round(Math.abs(thisValue - lastValue) / interval, 3).doubleValue();
            multiDataEntity.setCalValues(Tuple2.of(calValues.f0, roc));
        }
        out.collect(multiDataEntity);
    }

    /**
     * 更新缓存中存储的数据
     */
    private void cacheLastValue(MultiDataEntity entity) throws Exception {
        Integer targetId = entity.getTarget().getId();
        DateTime thisTime = DateUtil.date(entity.getPickTime() / 1000);
        Double thisValue = entity.getRawValues().get(referQuotaCode);
        if (lastValueMap.contains(targetId)) {
            Queue<Tuple2<DateTime, Double>> entityQueue = lastValueMap.get(targetId);
            entityQueue.offer(Tuple2.of(thisTime, thisValue));
            Optional
                    .ofNullable(entityQueue.peek())
                    .ifPresent(last -> {
                        DateTime lastTime = last.f0;
                        long diffTime = DtUtil.diffTime(thisTime, lastTime, timeUnit);
                        if (diffTime >= timeSpan) {
                            entityQueue.poll();
                        }
                    });
        } else {
            ArrayDeque<Tuple2<DateTime, Double>> adq = new ArrayDeque<>();
            adq.offer(Tuple2.of(thisTime, thisValue));
            lastValueMap.put(targetId, adq);
        }
    }

    /**
     * 获取上次的数据
     */
    private Double getLastValue(int targetId) throws Exception {
        Tuple2<DateTime, Double> lastTuple = lastValueMap.get(targetId).peek();
        if (null != lastTuple) {
            return lastTuple.f1;
        }
        return null;
    }

}
